package com.rrd.hive.udf;

import com.rrd.hive.util.PIDValidator;
import com.weicai.caesar.CaesarUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;


@Description(
        name = "pid_area",
        value = "_FUNC_(pid) - Get area string from pid or caesar encode pid",
        extended = "Example:\n> SELECT _FUNC_('110106201012310313');\n  北京市,丰台区,丰台区"
)
public class PidAreaUDF extends UDF {
    public static final Map<String, String> areaMap = new HashMap<>(255);
    public static String dataFile = "/data/dim/idno_area/idno_area";
    private static final   Map<String, String> cityMap = new HashMap<>(125);
    static {
        try {
            readAreaFile();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    public static Configuration config() {
        System.setProperty("HADOOP_USER_NAME", "hdfs");
        Configuration conf = new Configuration();
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        return conf;
    }

    public static void readAreaFile() throws IOException {
        Configuration conf = config();
        FileSystem fs = FileSystem.newInstance(URI.create(dataFile), conf);
        Path remotePath = new Path(dataFile);
        FSDataInputStream in = fs.open(remotePath);
        if (fs.exists(remotePath)) {
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in,"utf-8"));
            String lineTxt;
            String[] fieldValue;
            while ((lineTxt = bufferedReader.readLine()) != null) {
                fieldValue = lineTxt.split(",");
                areaMap.put(fieldValue[0], lineTxt.substring(fieldValue[0].length() + 1));
                cityMap.put(fieldValue[0].substring(0,4),fieldValue[1]+","+(fieldValue.length>=3?fieldValue[2]:""));
            }
            bufferedReader.close();
            in.close();
            fs.close();
        } else {
            System.out.println("找不到指定的文件");
        }
    }

    public Text evaluate(String pid) {
        if (pid == null || pid.equals("")) {
            return null;
        }
        Text text = new Text();
        if (CaesarUtil.isEncrypted(pid)) {
            pid = CaesarUtil.decode(pid);
        }
        if (!PIDValidator.isValidIDCard(pid)) {
            return null;
        }
        String area = areaMap.get(pid.substring(0, 6));

        if (area == null || area.equals("")) {
            area=cityMap.get(pid.substring(0, 4));
        }
        if (area != null) {
            text.set(area);
        }
        return text;
    }
    public static void main(String[] args) {
        PidAreaUDF dfu = new PidAreaUDF();
        System.out.println(dfu.evaluate("13811363008"));
    }

}
